package com.xiao.dynamic_rule.functions;

import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.pojo.ResultBean;
import com.xiao.dynamic_rule.pojo.RuleParam;
import com.xiao.dynamic_rule.service.QueryRouterV3;
import com.xiao.dynamic_rule.service.QueryRouterV4;
import com.xiao.dynamic_rule.utils.RuleSimulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 规则处理主逻辑
 */
public class RuleProcessFunctionV4 extends KeyedProcessFunction<String, LogBean, ResultBean> {
    private RuleParam ruleParam;
    private ListState<LogBean> listState;
    private QueryRouterV4 queryRouterV4;

    /**
     * 获取规则，并且设置2小时有效期的state
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {

        //获取查询路由对象
        queryRouterV4 = new QueryRouterV4();
        /**
         * 获取规则参数
         * TODO 模拟器生成的规则，后面要改成传入
         */
        ruleParam = RuleSimulator.getRuleParam();

        //创建一个存储数据的 listState
        ListStateDescriptor<LogBean> desc = new ListStateDescriptor<>("eventState", LogBean.class);
        //开启state的生命周期,存活俩小时
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).updateTtlOnCreateAndWrite().build();
        desc.enableTimeToLive(ttlConfig);
        listState = getRuntimeContext().getListState(desc);

    }

    /**
     * 2小时内的条件在state中查，2小时外的去clickhouse中查
     * @param logBean 传过来的封装成对象的数据
     * @param context flink上下文
     * @param out process的收集器，将处理的数据发往下游
     * @throws Exception
     */
    @Override
    public void processElement(LogBean logBean, Context context, Collector<ResultBean> out) throws Exception {

        //将数据放入state中
        listState.add(logBean);

        if (ruleParam.getTriggerParam().equals(logBean.getEventId())){
            //查询画像
            boolean userProfileResult = queryRouterV4.userProfileQuery(ruleParam, logBean);
            if (!userProfileResult) return;

            //查询行为次数条件
            boolean countResult = queryRouterV4.countConditionQuery(ruleParam, logBean, listState);
            if (!countResult) return;

            //查询行为序列条件
            boolean sequenceResult = queryRouterV4.sequenceConditionQuery(ruleParam, logBean, listState);
            if (!sequenceResult) return;

            // 说明都满足，返回成功结果集
            out.collect(new ResultBean(ruleParam.getRuleId(), logBean.getDeviceId(), logBean.getTimeStamp()));
        }
    }
}
